package com.we.risk.behavior.repay;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.math.BigDecimal;
import java.sql.Timestamp;

import static com.we.risk.behavior.repay.CommonVar.*;

public class TableRepayPlan {
    public static final String KFKTABLE = "fdm.uni01_repay_plan_230426";
//    public static final String KFKTABLE = "fdm.uni01_repay_plan";

    public static final String QUERYSQL =
            "select user_key as account_user_key, loan_key, "
                    + "periods, due_date, "
                    + "principal, " +
                    "payoff_time, create_time, update_time from "
                    + KFKTABLE +
                    " where user_key is not null " +
                    " and loan_key is not null " +
                    " and periods is not null " +
                    " and create_time is not null " +
                    " and update_time is not null " +
                    " and due_date is not null";
    public static Logger LOG = LoggerFactory.getLogger(TableRepayPlan.class);

    public String account_user_key;
    public String loan_key;
    public Integer periods;
    public BigDecimal principal;
    public Timestamp due_date;
    public Timestamp create_time;
    public Timestamp update_time;
    public Timestamp payoff_time;

    public TableRepayPlan() {}

    public TableRepayPlan(String account_user_key, String loan_key, Integer periods,
                          BigDecimal principal, Timestamp due_date, Timestamp create_time,
                          Timestamp update_time, Timestamp payoff_time) {
        this.account_user_key = account_user_key;
        this.loan_key = loan_key;
        this.periods = periods;
        this.principal = principal;
        this.due_date = due_date;
        this.create_time = create_time;
        this.update_time = update_time;
        this.payoff_time = payoff_time;
    }

    public static TableRepayPlan init(String account_user_key, String loan_key, Integer periods) {
        return new TableRepayPlan(account_user_key, loan_key, periods,
                null, null, null, null, null);
    }

    public static KeyedStream<String, String> getFlatMapStream(
            StreamTableEnvironment tableEnv, StreamExecutionEnvironment env) {
        Table tableRes = tableEnv.sqlQuery(TableRepayPlan.QUERYSQL);
        DataStream<Tuple2<Boolean, TableRepayPlan>> inputStream =
                tableEnv.toRetractStream(tableRes, TableRepayPlan.class);

        KeyedStream<String, String> out = inputStream.filter(new FilterFunction<Tuple2<Boolean, TableRepayPlan>>() {
            @Override
            public boolean filter(Tuple2<Boolean, TableRepayPlan> value) throws Exception {
                return value.f0;
            }
        }).uid("TableRepayPlan-filter")
//                .flatMap(new RichFlatMapFunction<Tuple2<Boolean, TableRepayPlan>, String>() {
//                    @Override
//                    public void flatMap(Tuple2<Boolean, TableRepayPlan> value, Collector<String> out) throws Exception {
//                        out.collect(value.f1.toString());
//                    }
//                }).keyBy(new KeySelector<String, String>() {
//                    @Override
//                    public String getKey(String value) throws Exception {
//                        JSONObject jsonObject = JSONObject.parseObject(value);
//                        return jsonObject.getString(LOANKEY);
//                    }
//                });
                .keyBy(new KeySelector<Tuple2<Boolean, TableRepayPlan>, String>() {
                    @Override
                    public String getKey(Tuple2<Boolean, TableRepayPlan> value) throws Exception {
                        return value.f1.getAccount_user_key() + value.f1.getLoan_key() + value.f1.getPeriods();
                    }
                }).flatMap(new RichFlatMapFunction<Tuple2<Boolean, TableRepayPlan>, String>() {
                    ValueState<TableRepayPlan> repayPlanVState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        repayPlanVState = getRuntimeContext()
                                .getState(new ValueStateDescriptor<TableRepayPlan>("TableRepayPlanVState", TableRepayPlan.class));
                    }

                    @Override
                    public void flatMap(Tuple2<Boolean, TableRepayPlan> value, Collector<String> out) throws Exception {
                        try {
                            TableRepayPlan repayPlan = value.f1;
                            String account_user_key = repayPlan.getAccount_user_key();
                            String loan_key = repayPlan.getLoan_key();
                            Integer periods = repayPlan.getPeriods();
                            Timestamp update_time = repayPlan.getUpdate_time();
                            TableRepayPlan tmpState = repayPlanVState.value();
                            TableRepayPlan curState = (tmpState == null ? init(account_user_key, loan_key, periods) : tmpState);

                            if (curState.update_time == null || update_time.getTime() > curState.update_time.getTime()) {
                                if(!curState.sameWith(repayPlan)) {
                                    if (repayPlan.loan_key != null) {
                                        out.collect(repayPlan.toString());
                                        repayPlanVState.update(repayPlan);
                                    }
                                }
                            }
                        } catch (Exception e) {
                            LOG.error("TableRepayPlan flatMap exec: " + e.toString());
                            LOG.error("TableRepayPlan flatMap exec input: " + value.toString());
                        }
                    }
                }).keyBy(new KeySelector<String, String>() {
                    @Override
                    public String getKey(String value) throws Exception {
                        JSONObject jsonObject = JSONObject.parseObject(value);
                        return jsonObject.getString(LOANKEY);
                    }
                });

        return out;
    }

    private boolean sameWith(TableRepayPlan plan) {
        boolean res = false;
        Timestamp due_date = plan.getDue_date();
        Timestamp payoff_time = plan.getPayoff_time();
        Timestamp create_time = plan.getCreate_time();
        BigDecimal principal = plan.getPrincipal();

        if(this.due_date == null || due_date == null) {
            res = (this.due_date == due_date);
            LOG.debug("curState.due_date: " + (this.due_date == null ? "null" : this.due_date) + " due_date: " + (due_date == null ? "null" : due_date));
        } else {
            res = (this.due_date.compareTo(due_date) == 0);
            LOG.debug("curState.due_date.compareTo(due_date): " + this.due_date.compareTo(due_date));
        }
        LOG.debug("sameWith res due_date: " + res);

        if(this.create_time == null || create_time == null) {
            res &= (this.create_time == create_time);
            LOG.debug("curState.create_time: " + (this.create_time == null ? "null" : this.create_time) + " create_time: " + (create_time == null ? "null" : create_time));
        } else {
            res &= (this.create_time.compareTo(create_time) == 0);
            LOG.debug("curState.create_time.compareTo(create_time): " + this.create_time.compareTo(create_time));
        }
        LOG.debug("sameWith res create_time: " + res);

        if(this.payoff_time == null || payoff_time == null) {
            res &= (this.payoff_time == payoff_time);
            LOG.debug("curState.payoff_time: " + (this.payoff_time == null ? "null" : this.payoff_time) + " payoff_time: " + (payoff_time == null ? "null" : payoff_time));
        } else {
            res &= (this.payoff_time.compareTo(payoff_time) == 0);
            LOG.debug("curState.payoff_time.compareTo(payoff_time): " + this.payoff_time.compareTo(payoff_time));
        }
        LOG.debug("sameWith res payoff_time: " + res);

        if(this.principal == null || principal == null) {
            if(this.principal == null && principal == null) {
                res &= true;
            } else {
                res &= false;
            }
            LOG.debug("curState.left_principal: " + (this.principal == null ? "null" : this.principal) + " left_principal: " + (principal == null ? "null" : principal));
        } else {
            res &= (this.principal.equals(principal));
            LOG.debug("curState.left_principal.equals(principal): " + this.principal.equals(principal));

        }
        LOG.debug("sameWith res left_principal: " + res);
        return res;
    }

    @Override
    public String toString() {
        JSONObject outJson = new JSONObject();
        outJson.put(ACCUSERKEY, account_user_key);
        outJson.put(LOANKEY, loan_key);
        outJson.put(PERIODS, periods);
        outJson.put("principal", principal);
        outJson.put(DUEDATE, due_date);
        outJson.put("create_time", create_time);
        outJson.put("update_time", update_time);
        outJson.put(PAYOFFTIME, payoff_time);
        return outJson.toString();
    }

    public String getAccount_user_key() {
        return account_user_key;
    }

    public String getLoan_key() {
        return loan_key;
    }

    public Integer getPeriods() {
        return periods;
    }

    public BigDecimal getPrincipal() {
        return principal;
    }

    public Timestamp getDue_date() {
        return due_date;
    }

    public Timestamp getCreate_time() {
        return create_time;
    }

    public Timestamp getUpdate_time() {
        return update_time;
    }

    public Timestamp getPayoff_time() {
        return payoff_time;
    }
}
